package com.doit.day03

import scala.collection.mutable
import scala.util.Random

object Test {
  def main(args: Array[String]): Unit = {
    val ints: Seq[Int] = Seq[Int](0, 1, 2, 3, 4, 5, 6, 7, 8, 9,10,11,12,13,14,15,16,17,18,19)
      val intToInts: mutable.Map[Int, Seq[Int]] = assignReplicasToBrokersRackUnaware(10, 8, ints, -1, -1)
      intToInts.toList.sortBy(_._1).foreach(println)
  }

 def assignReplicasToBrokersRackUnaware(nPartitions: Int, //分区的个数   10
                                        replicationFactor: Int,  //副本的个数  5
                                        brokerList: Seq[Int],//broker的集合
                                        fixedStartIndex: Int,//默认值是-1  固定开始的索引位置
                                        startPartitionId: Int): mutable.Map[Int, Seq[Int]] = {
    //创建了一个可变的map集合
    val ret = mutable.Map[Int, Seq[Int]]()

    //将broker的序列转换成了数组
    val brokerArray = brokerList.toArray

    //startIndex =4
    val startIndex = 18/*if (fixedStartIndex >= 0) {
      fixedStartIndex
    }else {
      Random.nextInt(brokerArray.length)
    }*/
   println("第一个随机数："+startIndex)

    var currentPartitionId = math.max(0, startPartitionId)

    var nextReplicaShift = 16/*if (fixedStartIndex >= 0) {
      fixedStartIndex
    }else {
      Random.nextInt(brokerArray.length)
    }*/
   println("第二个随机数："+nextReplicaShift)
    for (_ <- 0 until nPartitions) {
      if (currentPartitionId > 0 && (currentPartitionId % brokerArray.length == 0)){
        nextReplicaShift += 1
      }
      val firstReplicaIndex = (currentPartitionId + startIndex) % brokerArray.length

      val replicaBuffer = mutable.ArrayBuffer(brokerArray(firstReplicaIndex))

      for (j <- 0 until replicationFactor - 1){
        replicaBuffer += brokerArray(replicaIndex(firstReplicaIndex, nextReplicaShift, j, brokerArray.length))
      }

      ret.put(currentPartitionId, replicaBuffer)
      currentPartitionId += 1
    }
    ret
  }

  private def replicaIndex(firstReplicaIndex: Int, secondReplicaShift: Int, replicaIndex: Int, nBrokers: Int): Int = {
    val shift = 1 + (secondReplicaShift + replicaIndex) % (nBrokers - 1)
    (firstReplicaIndex + shift) % nBrokers
  }

}
